package com.ylkj.bus.device.report;

import com.alibaba.fastjson2.JSONObject;
import com.hwl.redis.RedisConfig;
import com.ylkj.bus.device.onoffline.model.OnOfflineDeviceEnty;
import com.ylkj.dao.RdsConfig;
import com.ylkj.dao.RdsUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.*;
import java.text.MessageFormat;
import java.time.*;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.Date;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 设备状态更新处理类，用于处理设备上下线事件并更新数据库状态
 * @Author: wengzx
 * @Date: 2025/4/22
 */
@Slf4j
public class UpdateDevicReportAndUnpack extends ProcessFunction<OnOfflineDeviceEnty, String> {

    // SQL模板定义
    private static final String BASE_UPDATE_OFFLINE_SQL1 = "UPDATE {0} SET roma_online_status = ?, last_office_time = ?, update_time = ? WHERE device_code = ? AND (last_online_time < ? OR last_online_time IS NULL)";
    private static final String BASE_UPDATE_OFFLINE_SQL2 = "UPDATE {0} SET last_office_time = ?, update_time = ? WHERE device_code = ? AND (last_office_time < ? OR last_office_time IS NULL)";
    private static final String BASE_UPDATE_ONLINE_SQL1 = "UPDATE {0} SET roma_online_status = ?, last_online_time = ?, update_time = ? WHERE device_code = ? AND (last_office_time < ? OR last_office_time IS NULL)";
    private static final String BASE_UPDATE_ONLINE_SQL2 = "UPDATE {0} SET last_online_time = ?, update_time = ? WHERE device_code = ? AND (last_online_time < ? OR last_online_time IS NULL)";
    //基础删除SQL模板
    private static final String DELETE_SQL_TEMPLATE = "DELETE FROM {0} WHERE report_data < ?";
    private static final String INSERT_SQL =
            "INSERT INTO {0} (`mid`, `device_code`, `device_status`, `report_data`,`event_time`,`update_num`, `create_time`, `update_time`) " +
                    "VALUES (?, ?, ?, ?, ?, ?, now(), now()) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`mid` = VALUES(`mid`), " +
                    "`device_code` = VALUES(`device_code`), " +
                    "`device_status` = VALUES(`device_status`), " +
                    "`report_data` = VALUES(`report_data`), " +
                    "`event_time` = VALUES(`event_time`), " +
                    "update_num = VALUES(`update_num`)";



    // 下线第1次更新
    private static final String BASE_MONITOR_OFFLINE_SQL1 =
            "INSERT INTO {0} (`device_code`, `alert_level`, `alert_dimensions`, `alarm_cover`, `event_type`, `event_value`, `alarm_time`, `blanking_tool`, `product_id`,`product_type`,`message_suggestion`,`alert_source`,`net_code`,`personnel_handler`,`handling_time`,`alert_description`,create_time, update_time) " +
                    "VALUES (?, ?, ?, ?, ?,  ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, now(), now()) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`alarm_time` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN VALUES(`alarm_time`) ELSE `alarm_time` END, " +
                    "`event_value` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN VALUES(`event_value`) ELSE `event_value` END, " +
                    "`alarm_cover` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN VALUES(`alarm_cover`) ELSE `alarm_cover` END, " +
                    "`update_time` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN now() ELSE `update_time` END";

    // 下线第2次更新
    private static final String BASE_MONITOR_OFFLINE_SQL2 =
            "INSERT INTO {0} (`device_code`, `alert_level`, `alert_dimensions`, `alarm_cover`, `event_type`, `event_value`, `alarm_time`, `blanking_tool`, `product_id`,`product_type`,`message_suggestion`,`alert_source`,`net_code`,`personnel_handler`,`handling_time`,`alert_description`,create_time, update_time) " +
                    "VALUES (?, ?, ?, ?, ?,  ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, now(), now()) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`alarm_time` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`alarm_time`) ELSE `alarm_time` END,  " +
                    "`event_value` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`event_value`) ELSE `event_value` END,  " +
                    "`update_time` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN now() ELSE `update_time` END";

    // 上线第1次更新
    private static final String BASE_MONITOR_ONLINE_SQL1 =
            "INSERT INTO {0} (`device_code`, `alert_level`, `alert_dimensions`, `alarm_cover`, `event_type`,   `blanking_tool`, `product_id`,`product_type`,`alert_source`,`net_code`,`handling_time`,create_time, update_time) " +
                    "VALUES (?, ?, ?, ?, ?,  ?, ?, ?, ?, ?, ?, now(), now()) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`alert_level` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`alert_level`) ELSE `alert_level` END,  " +
                    "`alert_dimensions` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`alert_dimensions`) ELSE `alert_dimensions` END,  " +
                    "`alarm_cover` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`alarm_cover`) ELSE `alarm_cover` END,  " +
                    "`handling_time` =  CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`handling_time`) ELSE `handling_time` END,  " +
                    "`update_time` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN now() ELSE `update_time` END";

    // 上线第2次更新
    private static final String BASE_MONITOR_ONLINE_SQL2 =
            "INSERT INTO {0} (`device_code`, `alert_level`, `alert_dimensions`, `alarm_cover`, `event_type`,   `blanking_tool`, `product_id`,`product_type`,`alert_source`,`net_code`,`handling_time`,create_time, update_time) " +
                    "VALUES (?, ?, ?, ?, ?,  ?, ?, ?, ?, ?, ?, now(), now()) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`handling_time` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN VALUES(`handling_time`) ELSE `handling_time` END,  " +
                    "`update_time` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN now() ELSE `update_time` END";

    // 数据库配置
    private final RdsConfig rdsConfig;
    private final String tableName;
    private final RedisConfig redisConfig;
    private RdsUtils rdsUtils;
    private final String tableName2;
    private final String tableName3;
    private Jedis jedis;
    // 格式化后的SQL语句
    private String baseOfflineSql1;
    private String baseOfflineSql2;
    private String baseOnlineSql1;
    private String baseOnlineSql2;
    private String baseInsertSql;

    private String baseMONITOROfflineSql1;
    private String baseMONITOROfflineSql2;
    private String baseMONITOROnlineSql1;
    private String baseMONITOROnlineSql2;
    // 死锁错误码定义
    private static final int DEADLOCK_ERROR_CODE = 1213;
    // 最大重试次数
    private static final int MAX_RETRY_TIMES = 3;
    // 定时任务相关
    private ScheduledExecutorService scheduler;


    public UpdateDevicReportAndUnpack(String rdsTableName, RdsConfig rdsConfig,RedisConfig redisConfig,String rdsTableName2,String rdsTableName3) {
        this.rdsConfig = rdsConfig;
        this.tableName2 = rdsTableName2;
        this.redisConfig = redisConfig;
        this.tableName = rdsTableName;
        this.tableName3 = rdsTableName3;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        rdsUtils = new RdsUtils(rdsConfig);
        jedis = redisConfig.getJedis();
        // 初始化SQL模板
        baseOfflineSql1 = MessageFormat.format(BASE_UPDATE_OFFLINE_SQL1, tableName);
        baseOfflineSql2 = MessageFormat.format(BASE_UPDATE_OFFLINE_SQL2, tableName);
        baseOnlineSql1 = MessageFormat.format(BASE_UPDATE_ONLINE_SQL1, tableName);
        baseOnlineSql2 = MessageFormat.format(BASE_UPDATE_ONLINE_SQL2, tableName);
        baseInsertSql = MessageFormat.format(INSERT_SQL, tableName2);
        // 初始化SQL模板
        baseMONITOROfflineSql1 = MessageFormat.format(BASE_MONITOR_OFFLINE_SQL1, tableName3);
        baseMONITOROfflineSql2 = MessageFormat.format(BASE_MONITOR_OFFLINE_SQL2, tableName3);
        baseMONITOROnlineSql1 = MessageFormat.format(BASE_MONITOR_ONLINE_SQL1, tableName3);
        baseMONITOROnlineSql2 = MessageFormat.format(BASE_MONITOR_ONLINE_SQL2, tableName3);
        // 创建定时任务调度器
        scheduler = Executors.newSingleThreadScheduledExecutor();
        // 立即执行一次清理任务
        deleteOldData();
        long initialDelay = calculateInitialDelay();
        scheduler.scheduleAtFixedRate(
                this::deleteOldData,
                initialDelay,
                TimeUnit.DAYS.toSeconds(1),
                TimeUnit.SECONDS
        );
        log.info("定时清理任务已启动，首次执行时间：{}秒后", initialDelay/1000);
    }



    @Override
    public void processElement(OnOfflineDeviceEnty values, Context ctx, Collector<String> out) throws Exception {
        if (values == null) {
            log.warn("收到空设备状态数据，跳过处理");
            return;
        }

        String deviceCode = values.getDeviceCode();
        Date updateTime = values.getUpdateTime();
        String romaOnlineStatus = values.getRomaOnlineStatus();
        Date lastOnlineStatusTime = values.getLastOnlineStatusTime();

        // 验证状态有效性
        if (!isValidStatus(romaOnlineStatus)) {
            log.warn("无效的设备状态值: {}", romaOnlineStatus);
            return;
        }
        int retryCount = 0;
        while (retryCount < MAX_RETRY_TIMES) {
            try (Connection connection = rdsUtils.getConnection()) {
                connection.setAutoCommit(false);

                int affectedRows = 0; // 将 affectedRows 定义为局部变量
                boolean updateResult = false;
                boolean updatemonitorResult = false;
                int[] affectedRowsHolder = new int[1]; // 在外部定义数组
                if ("1".equals(romaOnlineStatus)) {
                    updateResult = handleOnlineStatus(connection, deviceCode, updateTime, lastOnlineStatusTime, affectedRowsHolder);
                    updatemonitorResult = handlemonitorOnlineStatus(connection, deviceCode, values, affectedRowsHolder);
                    affectedRows = affectedRowsHolder[0]; // 在调用后获取值
                } else if ("2".equals(romaOnlineStatus)) {
                    updateResult = handleOfflineStatus(connection, deviceCode, updateTime, lastOnlineStatusTime, affectedRowsHolder);
                    updatemonitorResult = handlemonitorOfflineStatus(connection, deviceCode, values, affectedRowsHolder);
                    affectedRows = affectedRowsHolder[0]; // 在调用后获取值
                }

                String targetKey = redisConfig.getTargetKey();
                String deviceOnlineStatus = jedis.hget(targetKey, "deviceOnlineStatus");
                if (deviceOnlineStatus == null || deviceOnlineStatus.isEmpty()) {
                    log.warn("Redis配置缺失，跳过历史记录处理");
                } else {
                    try {
                        JSONObject config = JSONObject.parseObject(deviceOnlineStatus);
                        String configValue = config.getString("configValue");

                        if ("1".equals(configValue)) {
                            handleOnOfflineHistory(connection, deviceCode, romaOnlineStatus, lastOnlineStatusTime, affectedRows);
                        } else {
                            log.debug("设备状态记录功能已禁用（configValue={}）", configValue);
                        }
                    } catch (Exception e) {
                        log.error("Redis配置解析异常", e);
                    }
                }
                connection.commit();
                log.debug("设备[{}]状态更新成功，影响行数: {}", deviceCode, updateResult ? 1 : 0);
                return;
            } catch (SQLException e) {
                handleSQLException(e, retryCount);
                retryCount++;
            }
        }
        log.error("设备[{}]状态更新失败，超过最大重试次数", deviceCode);
    }




    /**
     * 处理设备上线状态更新
     * @param connection 数据库连接
     * @param deviceCode 设备编码
     * @param updateTime 更新时间
     * @param lastOnlineStatusTime 最后在线时间
     * @return 是否成功更新状态
     */
    private boolean handleOnlineStatus(Connection connection, String deviceCode, Date updateTime, Date lastOnlineStatusTime, int[] affectedRowsHolder) throws SQLException {
        boolean firstUpdate = executeOnlineUpdate1(connection, deviceCode, updateTime, lastOnlineStatusTime, affectedRowsHolder);
        if (!firstUpdate) {
            executeOnlineUpdate2(connection, deviceCode, updateTime, lastOnlineStatusTime);
        }
        return firstUpdate;
    }

    /**
     * 执行上线状态第一次更新（更新状态和最后在线时间）
     */
    private boolean executeOnlineUpdate1(Connection connection, String deviceCode, Date updateTime, Date lastOnlineStatusTime, int[] affectedRowsHolder) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseOnlineSql1)) {
            setCommonParameters(stmt, "1", lastOnlineStatusTime, deviceCode, updateTime);
            stmt.setTimestamp(5, convertToSqlDate(lastOnlineStatusTime)); // 修复：确保调用 convertToSqlDate
            affectedRowsHolder[0] = stmt.executeUpdate();
            return affectedRowsHolder[0] > 0;
        }
    }

    /**
     * 执行上线状态第二次更新（仅更新最后在线时间）
     */
    private void executeOnlineUpdate2(Connection connection, String deviceCode,
                                      Date updateTime, Date lastOnlineStatusTime) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseOnlineSql2)) {

            setCommonParameters(stmt, null, lastOnlineStatusTime, deviceCode, updateTime);
            stmt.setTimestamp(4, convertToSqlDate(lastOnlineStatusTime));
            stmt.executeUpdate();
        }
    }

    /**
     * 处理设备上线状态更新
     * @param connection 数据库连接
     * @param deviceCode 设备编码
     * @param values 实体
     * @param affectedRowsHolder 影响行数持有者
     * @return 是否成功更新状态
     */
    private boolean handlemonitorOnlineStatus(Connection connection, String deviceCode,
                                              OnOfflineDeviceEnty values, int[] affectedRowsHolder) throws SQLException {
        boolean firstUpdate = executemonitorOnlineUpdate1(connection, deviceCode, values, affectedRowsHolder);
        if (!firstUpdate) {
            executemonitorOnlineUpdate2(connection, deviceCode, values);
        }
        return firstUpdate;
    }



    /**
     * 执行上线状态第一次更新
     */
    private boolean executemonitorOnlineUpdate1(Connection connection, String deviceCode,
                                                OnOfflineDeviceEnty values, int[] affectedRowsHolder) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseMONITOROnlineSql1)) {
            setOnlineParameters(stmt, deviceCode, values);
            // 设置ON DUPLICATE KEY UPDATE部分的参数
            Timestamp handlingTimeStr = convertToSqlDate(values.getHandlingTime());
            stmt.setTimestamp(12, handlingTimeStr);
            stmt.setTimestamp(13, handlingTimeStr);
            stmt.setTimestamp(14, handlingTimeStr);
            stmt.setTimestamp(15, handlingTimeStr);
            stmt.setTimestamp(16, handlingTimeStr);
            affectedRowsHolder[0] = stmt.executeUpdate();
            return affectedRowsHolder[0] > 0;
        }
    }

    /**
     * 执行上线状态第二次更新
     */
    private void executemonitorOnlineUpdate2(Connection connection, String deviceCode,
                                             OnOfflineDeviceEnty values) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseMONITOROnlineSql2)) {
            setOnlineParameters(stmt, deviceCode, values);
            // 设置ON DUPLICATE KEY UPDATE部分的参数
            Timestamp handlingTimeStr = convertToSqlDate(values.getHandlingTime());
            stmt.setTimestamp(12, handlingTimeStr);
            stmt.setTimestamp(13, handlingTimeStr);
            stmt.executeUpdate();
        }
    }

    /**
     * 处理设备下线状态更新
     * @param connection 数据库连接
     * @param deviceCode 设备编码
     * @param values 实体
     * @param affectedRowsHolder 影响行数持有者
     * @return 是否成功更新状态
     */
    private boolean handlemonitorOfflineStatus(Connection connection, String deviceCode,
                                        OnOfflineDeviceEnty values, int[] affectedRowsHolder) throws SQLException {
        boolean firstUpdate = executemonitorOfflineUpdate1(connection, deviceCode, values, affectedRowsHolder);
        if (!firstUpdate) {
            executemonitorOfflineUpdate2(connection, deviceCode, values);
        }
        return firstUpdate;
    }




    /**
     * 执行下线状态第一次更新
     */
    private boolean executemonitorOfflineUpdate1(Connection connection, String deviceCode,
                                                 OnOfflineDeviceEnty values, int[] affectedRowsHolder) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseMONITOROfflineSql1)) {
            setOfflineParameters(stmt, deviceCode, values);
            // 设置ON DUPLICATE KEY UPDATE部分的参数
            Timestamp alarmTimeStr = convertToSqlDate(values.getAlarmTime());
            stmt.setTimestamp(17, alarmTimeStr);
            stmt.setTimestamp(18, alarmTimeStr);
            stmt.setTimestamp(19, alarmTimeStr);
            stmt.setTimestamp(20, alarmTimeStr);
            affectedRowsHolder[0] = stmt.executeUpdate();
            return affectedRowsHolder[0] > 0;
        }
    }

    /**
     * 执行下线状态第二次更新
     */
    private void executemonitorOfflineUpdate2(Connection connection, String deviceCode,
                                              OnOfflineDeviceEnty values) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseMONITOROfflineSql2)) {
            setOfflineParameters(stmt, deviceCode, values);
            // 设置ON DUPLICATE KEY UPDATE部分的参数
            Timestamp alarmTimeStr = convertToSqlDate(values.getAlarmTime());
            stmt.setTimestamp(17, alarmTimeStr);
            stmt.setTimestamp(18, alarmTimeStr);
            stmt.setTimestamp(19, alarmTimeStr);
            stmt.executeUpdate();
        }
    }


    /**
     * 设置上线参数
     */
    private void setOnlineParameters(PreparedStatement stmt, String deviceCode,
                                     OnOfflineDeviceEnty values) throws SQLException {
        int paramIndex = 1;
        stmt.setString(paramIndex++, deviceCode);
        stmt.setString(paramIndex++, values.getAlertLevel());
        stmt.setString(paramIndex++, values.getAlertDimensions());
        stmt.setString(paramIndex++, values.getAlarmCover());
        stmt.setString(paramIndex++, values.getEventType());
        stmt.setString(paramIndex++, values.getBlankingTool());
        // 处理productId的null值
        Long productId = values.getProductId();
        if (productId != null) {
            stmt.setLong(paramIndex++, productId);
        } else {
            stmt.setNull(paramIndex++, Types.BIGINT); // 假设数据库字段为BIGINT类型
        }
        // 处理productType的null值
        String productType = values.getProductType();
        if (productType != null) {
            stmt.setString(paramIndex++, productType);
        } else {
            stmt.setNull(paramIndex++, Types.VARCHAR);
        }
        stmt.setString(paramIndex++, values.getAlertSource());
        stmt.setString(paramIndex++, values.getNetCode());
        stmt.setTimestamp(paramIndex++, convertToSqlDate(values.getHandlingTime()));
    }

    /**
     * 设置下线参数
     */
    private void setOfflineParameters(PreparedStatement stmt, String deviceCode,
                                      OnOfflineDeviceEnty values) throws SQLException {
        int paramIndex = 1;
        stmt.setString(paramIndex++, deviceCode);
        stmt.setString(paramIndex++, values.getAlertLevel());
        stmt.setString(paramIndex++, values.getAlertDimensions());
        stmt.setString(paramIndex++, values.getAlarmCover());
        stmt.setString(paramIndex++, values.getEventType());
        stmt.setString(paramIndex++, values.getEventValue());
        stmt.setTimestamp(paramIndex++, convertToSqlDate(values.getAlarmTime()));
        stmt.setString(paramIndex++, values.getBlankingTool());
        // 处理productId的null值
        Long productId = values.getProductId();
        if (productId != null) {
            stmt.setLong(paramIndex++, productId);
        } else {
            stmt.setNull(paramIndex++, Types.BIGINT); // 假设数据库字段为BIGINT类型
        }
        // 处理productType的null值
        String productType = values.getProductType();
        if (productType != null) {
            stmt.setString(paramIndex++, productType);
        } else {
            stmt.setNull(paramIndex++, Types.VARCHAR);
        }
        stmt.setString(paramIndex++, values.getMessageSuggestion());
        stmt.setString(paramIndex++, values.getAlertSource());
        stmt.setString(paramIndex++, values.getNetCode());
        stmt.setString(paramIndex++, values.getPersonnelHandler());
        stmt.setTimestamp(paramIndex++, convertToSqlDate(values.getHandlingTime()));
        stmt.setString(paramIndex++, values.getAlertDescription());
    }

    /**
     * 处理设备下线状态更新
     */
    private boolean handleOfflineStatus(Connection connection, String deviceCode, Date updateTime, Date lastOnlineStatusTime, int[] affectedRowsHolder) throws SQLException {
        boolean firstUpdate = executeOfflineUpdate1(connection, deviceCode, updateTime, lastOnlineStatusTime, affectedRowsHolder);
        if (!firstUpdate) {
            executeOfflineUpdate2(connection, deviceCode, updateTime, lastOnlineStatusTime);
        }
        return firstUpdate;
    }

    /**
     * 执行下线状态第一次更新（更新状态和最后离线时间）
     */
    private boolean executeOfflineUpdate1(Connection connection, String deviceCode,
                                          Date updateTime, Date lastOnlineStatusTime, int[] affectedRowsHolder) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseOfflineSql1)) {
            setCommonParameters(stmt, "2", lastOnlineStatusTime, deviceCode, updateTime);
            stmt.setTimestamp(5, convertToSqlDate((lastOnlineStatusTime)));
            affectedRowsHolder[0] = stmt.executeUpdate();
            return affectedRowsHolder[0] > 0;
        }
    }

    /**
     * 执行下线状态第二次更新（仅更新最后离线时间）
     */
    private void executeOfflineUpdate2(Connection connection, String deviceCode,
                                       Date updateTime, Date lastOnlineStatusTime) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseOfflineSql2)) {
            setCommonParameters(stmt, null, lastOnlineStatusTime, deviceCode, updateTime);
            stmt.setTimestamp(4, convertToSqlDate((lastOnlineStatusTime)));
            stmt.executeUpdate();
        }
    }
    /**
     * 设置公共参数到PreparedStatement
     */
    private void setCommonParameters(PreparedStatement stmt, String status, Date lastStatusTime, String deviceCode, Date updateTime) throws SQLException {
        int paramIndex = 1;
        if (status != null) {
            stmt.setString(paramIndex++, status);
        }
        stmt.setTimestamp(paramIndex++, convertToSqlDate(lastStatusTime)); // 修复：使用 setTimestamp 替代 setDate
        stmt.setTimestamp(paramIndex++, convertToSqlDate(updateTime));     // 修复：使用 setTimestamp 替代 setDate
        stmt.setString(paramIndex++, deviceCode);
    }

    private void handleOnOfflineHistory(Connection connection, String deviceCode, String romaOnlineStatus, Date lastOnlineStatusTime, int affectedRows) throws SQLException {
        try {
            // 生成UUID
            String uniqueId = UUID.randomUUID().toString();

            // 转换时间戳为字符串
            LocalDateTime dateTime = lastOnlineStatusTime.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
            String formattedDate = dateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            String eventTimeStr = String.valueOf(lastOnlineStatusTime.getTime()); // 保持时间戳字符串形式

            try (PreparedStatement stmt = connection.prepareStatement(baseInsertSql)) {
                stmt.setString(1, uniqueId); // 使用UUID
                stmt.setString(2, deviceCode);
                stmt.setString(3, romaOnlineStatus);
                stmt.setString(4, formattedDate.substring(0, 10)); // 提取日期部分
                stmt.setString(5, eventTimeStr); // 存储时间戳字符串
                stmt.setInt(6, affectedRows);
                stmt.executeUpdate();
            }
        } catch (Exception e) {
            log.error("处理历史记录失败，设备代码：{}，错误信息：{}", deviceCode, e.getMessage(), e);
        }
    }
    /**
     * 处理SQL异常
     */
    private void handleSQLException(SQLException e, int retryCount) throws SQLException {
        if (e.getErrorCode() == DEADLOCK_ERROR_CODE && retryCount < MAX_RETRY_TIMES) {
            log.warn("死锁检测，正在重试（第{}次）... 错误信息: {}", retryCount + 1, e.getMessage());
        } else {
            log.error("数据库操作失败，错误信息: {}", e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 验证状态值有效性
     */
    private boolean isValidStatus(String status) {
        return "1".equals(status) || "2".equals(status);
    }

    // 计算到当天凌晨的延迟时间（毫秒）
    private long calculateInitialDelay() {
        LocalDateTime nextRun = LocalDateTime.of(
                LocalDateTime.now().toLocalDate(),
                LocalTime.MIDNIGHT
        );

        if (LocalDateTime.now().isAfter(nextRun)) {
            nextRun = nextRun.plusDays(1);
        }
        return Duration.between(LocalDateTime.now(), nextRun).toMillis();
    }

    /**
     * 定时清理三天前的旧数据
     */
    private void deleteOldData() {
        try (Connection connection = rdsUtils.getConnection();
             PreparedStatement stmt = connection.prepareStatement(
                     MessageFormat.format(DELETE_SQL_TEMPLATE, tableName2))) {

            // 计算清理时间（当前时间三天前）
            LocalDate threeDaysAgo = LocalDate.now().minusDays(3); // 使用 LocalDate 而不是 LocalDateTime
            stmt.setDate(1, java.sql.Date.valueOf(threeDaysAgo)); // 显式设置为 java.sql.Date

            int rowsDeleted = stmt.executeUpdate();
            log.info("已清理{}条超过3天的旧数据", rowsDeleted);
        } catch (Exception e) {
            log.error("数据清理失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 将 java.util.Date 转换为 java.sql.Timestamp
     */
    private Timestamp convertToSqlDate(java.util.Date date) {
        if (date != null) {
            return new Timestamp(date.getTime());
        }
        return null;
    }

    @Override
    public void close() throws Exception {
        if (rdsUtils != null) {
            rdsUtils.close();
        }
        if (jedis != null) {
            jedis.close();
        }
        super.close();
    }
}